Codetopia 海港音樂祭，進場倒數 45 分鐘。天空的臉色比舞台總監還難看，厚重的雲層下壓，雷達圖上那條刺眼的強降雨帶，正像個不請自來的搖滾巨星，直奔會場而來。
調度中心裡，氣氛比外面更加凝重。Rhea｜道路維修隊領班 一邊緊盯著螢幕上閃爍的封路點位，一邊用肩膀夾著話機，聽著 Liam｜調度員 從另一頭傳來的緊急指令：「Rhea！A 區原定 18:00 的封路計畫有變，改成『先開後封』！還有，B 區如果下雨，所有作業『緊急暫停』優先！」
Rhea 感覺自己的血管快要跟市中心的交通一樣打結了。麻煩的不是指令，而是執行。各個團隊的系統裡，描述狀態的變數簡直是個聯合國——工程隊的系統用 status="active"，交管隊用兩個布林值 isPaused 和 isCleared 來組合，更離譜的是，有些外包團隊直接用純文字備註來記錄狀態！
這意味著，一個簡單的「暫停」，Rhea 需要登入四個不同的系統，修改天曉得多少個欄位，還常常因為漏掉某個，導致出現「半套生效」的災難：路障放下了，但通知沒發出去；監控關了，但警示燈還亮著。
「夠了！」Rhea 對著話機怒吼，「我不管你們後台怎麼搞，我只要一個會說話的紅綠燈！告訴我現在到底是準備中、已生效、已解除，還是緊急暫停！就這麼簡單！」
是的，總設計師，這就是我們今天的痛點：
status、isPaused）和語義散落在各處，沒有單一事實來源 (SSOT)，導致狀態切換的副作用（封路、發通知、開關監控）難以控制。if/else 的無底洞：程式碼裡充滿了 if status == 'active' and not isPaused: 這樣的 logique，每當出現一個新情境（像是豪雨警報、貴賓車隊通行），就得在層層疊疊的 if/else 裡再開一條分支。🧭 術語卡（今日會用到）
GoF｜State：允許一個物件在其內部狀態改變時改變它的行為。物件看起來似乎修改了它的類別。核心是將與特定狀態相關的行為局部化，並將不同狀態的行為放在不同的物件中。
EIP/EDA｜Event-Sourced State Machine / Process Manager：狀態不再是資料庫裡的一個欄位，而是由一系列的「事件」推導出來的結果。狀態的每一次改變，都會發布一個或多個副作用（例如發布一個 Command）。
MAS｜TrafficControlAgent + DF：在多代理系統中，一個交管代理 (Agent) 會向黃頁服務 (Directory Facilitator) 註冊自己擁有「號誌控制」的能力。狀態機則是這個代理人用來管理自身內部行為與規則的「內在秩序」。
讓我們把鏡頭倒回到 Rhea 崩潰前十分鐘，看看 Bruno｜現場指揮副手 試圖用現有系統處理「豪雨警報」時，後台那段充滿「壞味道」的程式碼長什麼樣子：
# 反例：把狀態當成一堆旗標，副作用散落在各處，到處都是 if/else
def apply_change(area, event):
# 狀況一：區域正在生效中
if area.status == "active":
if event == "rain_alert": # 如果收到豪雨警報
# 副作用 1: 手動呼叫廣播 API
api.broadcast(f"{area.name} 因豪雨暫停入場")
# 狀態變更 1: 修改 isPaused 旗標
area.isPaused = True
# 副作用 2: 如果舞台有電，還要記得斷電...這誰寫的？
if area.hasPower:
power.release("stage-a") # 這段邏輯為什麼會在這裡？紀錄在哪？
# 狀況二：區域還在準備中
elif area.status == "preparing":
if event == "go_live":
# 副作用 3: 直接呼叫道路 API
api.close_road(area) # 又是一段直接呼叫底層 Receiver 的程式碼
# 狀態變更 2: 修改 status 字串
area.status = "active"
# ...底下還有 10 種不同的例外情境，工程師只能靠著字串備註來補洞...
壞味道在哪裡？ 👃
狀態＝拼裝車：狀態是由一堆不相關的字串和布林值拼湊出來的，根本沒有「單一真相」。
轉移邏輯滿天飛：狀態轉換的 logique 分散在多個函式中，入口 (entry) 和出口 (exit) 的副作用（像是開始/停止監控）更是隨處可見，完全無法追蹤。
根本測不了：給定同樣的輸入事件，這個函式產生的副作用順序和結果可能完全不同， 因為它依賴太多外部世界的狀態 (area.hasPower)，且缺乏冪等性。
好了，吐槽完畢。讓我們聽聽 Nadia｜交管規則官 的專業建議。她拿出一張升級後的設計圖，說：「Rhea，妳要的號誌機，不只要會說話，還要能應付突發狀況並留下證據。我們不僅要『物件化』狀態，更要讓狀態轉移本身成為一個可審計、可回溯的『微型交易』。」
一句話解釋 State 模式 (升級版)：
把每個狀態封裝成獨立物件，由它決定事件是否合法、應轉移至何處。更關鍵的是，我們引入了「暫存前一狀態 (Memento)」的機制，確保 EmergencyPaused 能安全返回任何被打斷的狀態。每一次狀態轉換，都將被視為一次原子操作，確保副作用的冪等性與可追溯性。
Nadia 建議的狀態集合如下：
Preparing (準備中)
Active (已生效)
Cleared (已解除)
EmergencyPaused (緊急暫停)
其中，主要的作業狀態（如 Preparing, Active）都可以因為外部事件（如豪雨）進入 EmergencyPaused，並在狀況解除後，Resume 回到它原本的狀態。
設計約束：為確保可預測性，此處的狀態物件 (
Preparing,Active等) 應為無狀態的 (Stateless)。它們不應包含內部可變數據；所有需要的上下文都由TrafficControlContext提供。
✅ 狀態有限且互斥：當一個物件的行為取決於其狀態，且狀態的數量是有限的、彼此互斥的（例如，一個路口不可能是「綠燈」又是「紅燈」）。
✅ 轉移帶有副作用：狀態轉換時需要執行固定的進入 (entry) 或離開 (exit) 動作，例如從 Preparing 轉到 Active 時，必須 執行「封路」和「開啟監控」。State 模式可以確保這些動作被穩定執行。
✅ 需要一致的審計：當你需要清楚記錄每一次狀態轉換的原因和結果時，將轉移動作集中在狀態物件內，可以方便地加入日誌或事件發布。
✅ 想消除 if/else 或 switch：當你的程式碼中充斥著根據物件狀態來決定行為的龐大條件分支時。
⛔ 流程骨架固定，僅替換演算法：如果你的情境更像是一個固定的演算法流程，只是其中某個步驟需要替換不同實作，那更適合用 Template Method 模式（我們明天談！）。
⛔ 只是單純的策略切換：如果狀態之間沒有固定的轉移規則，隨時可以任意切換，且沒有 entry/exit 副作用，那麼更簡單的 Strategy 模式可能就足夠了。
⛔ 狀態數量極多或無限：如果狀態的數量非常龐大或動態增長，為每個狀態建立一個類別會導致類別爆炸，此時應考慮其他方法（例如，將狀態規則儲存在資料庫中）。
好的，導播，鏡頭拉一下！讓我們從三個不同的尺度，來看看 Nadia 設計的這套「交管號誌」系統是如何運作的。
如何閱讀：先看微觀的類別圖，理解 State 和 Context 的基本結構；再看中觀的流程圖，觀察事件如何驅動狀態轉移並產生副作用；最後在宏觀層面，理解這個狀態機如何作為一個代理 (Agent) 的「內在秩序」。
| 層級 | 對應概念 | Codetopia 詞彙 |
|---|---|---|
| 微觀 (GoF) | State / Context | 交管狀態物件 / 交管情境 |
| 中觀 (EIP/EDA) | Event-Sourced State Machine | 事件驅動的狀態轉移與副作用 Command |
| 宏觀 (MAS) | Agent's Internal FSM | 交管代理 (TrafficControlAgent) 的內在行為規則 |
這張圖展示了 State 模式的核心結構。TrafficControlContext (交管情境) 持有當前的狀態物件，並將所有事件處理都委派給它。EmergencyPaused 現在會記住是從哪個狀態來的。
這張圖描繪了實際的運作流程。主要的作業狀態(Preparing 和 Active)都能夠進入 EmergencyPaused，並且都能安全返回。而已解除狀態(Cleared)則會拒絕無意義的暫停請求。
光說不練假把戲。這是一段升級後的 Python 風格 pseudo code，它引入了版本控制、冪等命令、事務性轉移以及安全的暫停與恢復機制。
import uuid
# 模擬基礎設施
class CommandBus:
def send(self, cmd): print(f"✅ CMD SENT -> {cmd}")
class AuditLog:
def emit(self, event, **details): print(f"📝 AUDIT: {event} | {details}")
bus = CommandBus()
audit = AuditLog()
# --- 狀態介面與具體實作 ---
class State:
def entry(self, ctx): pass
def exit(self, ctx): pass
def on_event(self, ctx, event):
reason = f"Illegal event '{event}' in state '{self.__class__.__name__}'"
audit.emit("IllegalTransitionTried", reason=reason, area=ctx.area_name, version=ctx.version)
return {"status": "rejected", "reason": reason, "version": ctx.version}
class Preparing(State):
def entry(self, ctx): ctx.emit("Notify", {"msg": "進入準備中"})
def on_event(self, ctx, event):
if event == "go_live": return ctx.set_state(Active(), cause=event)
if event == "cancel": return ctx.set_state(Cleared(), cause=event)
if event == "rain_alert": return pause(ctx, cause=event)
return super().on_event(ctx, event)
class Active(State):
def entry(self, ctx):
# 進入生效：先確保封路，再開監控（冪等由 idempotencyKey 保護）
ctx.emit("CloseRoad", {})
ctx.emit("StartMonitor", {})
def exit(self, ctx): ctx.emit("StopMonitor", {})
def on_event(self, ctx, event):
if event == "rain_alert": return pause(ctx, cause=event)
if event == "clear": return ctx.set_state(Cleared(), cause=event)
return super().on_event(ctx, event)
class EmergencyPaused(State):
def __init__(self, resume_to: State): self.resume_to = resume_to
def entry(self, ctx): ctx.emit("Broadcast", {"msg": "因緊急狀況暫停"})
def on_event(self, ctx, event):
if event == "resume": return ctx.set_state(self.resume_to, cause=event)
return super().on_event(ctx, event)
class Cleared(State):
def entry(self, ctx):
ctx.emit("OpenRoad", {})
ctx.emit("RetractNotice", {})
def on_event(self, ctx, event):
if event == "rain_alert":
return {"status": "rejected", "reason": "no-op: already cleared", "version": ctx.version}
return super().on_event(ctx, event)
# --- 核心 Context 與輔助函式 ---
class TrafficControlContext:
def __init__(self, area_name):
self.area_name = area_name
self.version = 0
self.state = None
self.current_transition_id = None
self.set_state(Preparing(), cause="initialization")
def set_state(self, new_state: State, cause: str):
# 使用完整 uuid hex 降低碰撞機率（仍可讀）
self.current_transition_id = uuid.uuid4().hex
old_state_name = self.state.__class__.__name__ if self.state else "None"
audit.emit("TransitionStarted", id=self.current_transition_id, from_=old_state_name, to=new_state.__class__.__name__, cause=cause)
try:
if self.state: self.state.exit(self)
self.state = new_state
self.state.entry(self)
except Exception as e:
audit.emit("TransitionFailed", id=self.current_transition_id, reason=str(e))
# 這裡可以加入更複雜的回滾邏輯
return {"status": "error", "reason": "transition failed"}
self.version += 1
audit.emit("TransitionCommitted", id=self.current_transition_id, to=self.state.__class__.__name__, new_version=self.version)
return {"status": "handled", "new_state": self.state.__class__.__name__, "version": self.version}
def emit(self, type_, payload):
command = {
"type": type_,
"payload": {"area": self.area_name, **payload},
"idempotencyKey": f"{self.current_transition_id}:{type_}",
"correlationId": self.current_transition_id
}
bus.send(command)
def handle(self, event, expected_version):
print(f"\n⚡️ EVENT RECEIVED: '{event}' for {self.area_name}")
if self.version != expected_version:
reason = f"Version mismatch: expected {expected_version}, but is {self.version}"
audit.emit(
"ConcurrencyConflict",
reason=reason,
area=self.area_name,
event=event,
expected=expected_version,
actual=self.version,
state=self.state.__class__.__name__,
)
return {"status": "rejected", "reason": reason, "version": self.version}
return self.state.on_event(self, event)
def pause(ctx, cause):
if isinstance(ctx.state, EmergencyPaused):
return {"status": "rejected", "reason": "already_paused"}
return ctx.set_state(EmergencyPaused(resume_to=ctx.state), cause=cause)
# --- 讓我們在現場實際操作看看 (帶版本號) ---
print("--- 🎬 Harbor-A 交管啟動 (V3) ---")
tc = TrafficControlContext("Harbor-A")
current_version = tc.version
res = tc.handle("go_live", current_version)
current_version = res.get("version", current_version)
print("\n-- 從 Active 暫停 --")
res = tc.handle("rain_alert", current_version)
current_version = res.get("version", current_version)
print("\n-- 恢復到 Active --")
res = tc.handle("resume", current_version)
current_version = res.get("version", current_version)
當你在程式碼中看到以下信號時，就該警惕可能需要 State 模式來拯救了：
🚩 布林地獄 (Boolean Hell)：用一長串 isActive && !isPaused || isCleared 這樣的布林旗標組合來判斷物件狀態。
🚩 副作用四散：封路、發通知的程式碼散落在 UI 層、服務層、甚至排程腳本中，沒有統一的 entry/exit 約定。
🚩 狀態洩漏 (State Leakage)：外部程式碼可以直接修改狀態，例如 tc.state = "cleared"，完全繞過了狀態轉移的守門人邏輯。
🚩 無審計能力：沒有任何事件或狀態快照的紀錄，當系統出錯時，完全無法回溯（Replay）當時發生了什麼事。
State 模式不僅僅是 GoF 的一個章節，它更是通往更宏大架構的基石：
升維到 EDA (事件驅動架構)：在更嚴謹的系統中，可以將「決策」與「副作用」徹底分離。狀態物件的 on_event 方法可以變成一個純函數 (state, event) -> (new_state, commands[])，只回傳新的狀態和待執行的命令列表。由 Context 負責派送這些命令，這讓測試變得極其簡單，也為事件溯源 (Event Sourcing) 鋪平了道路。
併發控制與可靠投遞：如範例所示，引入 version 進行樂觀併發控制，是應對多事件源的標準做法。此外，所有派發的 Command 都應包含 idempotencyKey 和 correlationId，並結合 Outbox 模式，確保命令的投遞是可靠且可重入的 (Exactly-once delivery)。
與 Mediator (協調者) 串接：FestivalMediator (協調中心) 只需下達高階意圖事件。TrafficControlAgent 內部的狀態機，會根據自己當前的狀態和版本，決定這個事件是否合法，以及應該觸發哪些補償命令。
現在，讓我們用 Nadia 的升級版設計，重新跑一遍開場時的災難場景：
17:50：Bruno 啟動 A 區交管，系統狀態進入 Preparing (version 1)。
📝 AUDIT: TransitionCommitted | {'id': ..., 'to': 'Preparing', 'new_version': 1}
18:00：Bruno 按下「生效」，系統收到 go_live (expected_version: 1)。
📝 AUDIT: TransitionCommitted | {'id': ..., 'to': 'Active', 'new_version': 2}
18:15：雷達發出 rain_alert (expected_version: 2)。
📝 AUDIT: TransitionCommitted | {'id': ..., 'to': 'EmergencyPaused', 'new_version': 3}
18:16：(狀況) 調度中心 Liam 因網路延遲，送出一個基於舊狀態的 clear 指令 (expected_version: 2)。
📝 AUDIT: ConcurrencyConflict | {'reason': 'Version mismatch: expected 2, but is 3'}
系統拒絕了該操作，避免了數據不一致。
18:30：雨勢轉弱，Rhea 按下「恢復」，系統收到 resume (expected_version: 3)。
📝 AUDIT: TransitionCommitted | {'id': ..., 'to': 'Active', 'new_version': 4}
系統正確地回到了 Active 狀態。
驗收標準：
即時性：任一事件到達後，狀態切換與對應的 Command 派發在 2 秒內完成。(通過)
可追溯性：全流程產生可回放的審計日誌。(通過)
安全性：不允許非法轉移，且能處理版本衝突。(通過)
Rhea 和 Liam 都鬆了一口氣。這個新系統不僅會說話，還很聰明，不會被混亂的指令搞糊塗。
要確保這套「號誌機」永不出錯，我們可以從這幾個角度進行測試：
狀態轉移表測試：建立一個表格，列舉所有 (from_state, on_event) -> to_state 的組合。使用參數化測試，確保每一條合法的轉移路徑都如預期般運作。
Entry/Exit 契約測試：使用一個假的 (Mock) CommandBus，驗證在狀態轉換時，對應的 entry 和 exit 動作被正確、且依序地呼叫。
轉移閉包性 (Closure) 測試：對每個狀態，窮舉所有可能的事件，斷言結果要嘛是合法的轉移，要嘛是明確的 rejected 回應，絕不允許靜默地吞掉事件。
併發穩健性 (Robustness) 測試：模擬兩個事件（如 rain_alert 和 clear）以交錯、亂序的方式抵達，驗證系統能透過版本號正確處理，只接受其中一個，並拒絕另一個。
暫停恢復路徑測試：驗證 Preparing -> rain_alert -> resume -> go_live 的完整路徑，確保從「準備中」暫停後，能正確恢復到「準備中」，並繼續後續流程。
總設計師，輪到你了！來動手挑戰看看吧：
擴充 PartialActive 狀態：如果音樂祭需要一個「局部生效」的狀態（例如，只開放單線道通行），你會如何新增這個 Partial-Active 狀態？它應該有哪些專屬的 entry/exit 動作（例如，發出「部分路段開放，請小心駕駛」的通知）？
雨勢震盪挑戰：想像一下，如果氣象系統在 10 秒內連續發送了 rain_alert -> resume -> rain_alert...你會如何引入一個「去抖動 (debounce)」或「抑制器 (Inhibitor)」機制？（提示：也許可以引入一個 WeatherHolding 的中介狀態？）
小投票：副作用放哪裡？ 你會把觸發副作用 (例如 ctx.emit(...)) 的程式碼放在：
A. entry/exit 方法裡
B. on_event 方法裡（在 ctx.set_state(...) 之前）
你選擇的理由是什麼？這兩種做法在可維護性和可追溯性上有何差異？
一句話總結：把臨時交管號誌化——讓狀態自己說話、轉移成為守門人、副作用各歸其位，最終讓一切審計都可回放。
今天，我們用 State 模式把混亂的旗標變成了可靠的號誌機。但你可能會想，這些 entry、exit 的呼叫流程，是不是有點重複？有沒有辦法把它們變成一個更穩定的「骨架」呢？
明日預告：Day 21｜Template Method（骨架＋鉤子）—— 我們將把今天的狀態機規範，提升為一個「固定骨架＋可覆寫鉤子」的穩定模板，打造 Codetopia 的標準作業流程 (SOP)。
為了確保在不支援 Mermaid 渲染的環境中也能正常閱讀，以下提供文中圖表的 ASCII 替代版本：
┌─────────────────────────────────┐
│ TrafficControlContext │
├─────────────────────────────────┤
│ - state: State │
│ - version: int │
│ - current_transition_id: string │
├─────────────────────────────────┤
│ + handle(event, expected_ver) │
│ + set_state(State, cause) │
│ + emit(type, payload) │
└─────────────────┬───────────────┘
│ delegates
│
▼
┌─────────────────┐
│ State │
│ <<interface>> │
├─────────────────┤
│ + on_event(ctx, event)
│ + entry(ctx) │
│ + exit(ctx) │
└─────────┬───────┘
│
┌──────────┼──────────┐
│ │ │
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│Preparing│ │ Active │ │ Cleared │
└─────────┘ └─────────┘ └─────────┘
│
▼
┌─────────────────────┐
│ EmergencyPaused │
├─────────────────────┤
│ - resume_to: State │
└─────────────────────┘
Traffic Control FSM
╔══════════════════════════════════════════════════════════╗
║ ║
║ [外部事件] ║
║ │ ║
║ ▼ ║
║ ┌─────────────┐ go_live ┌─────────────┐ ║
║ │ Preparing ├────────────────────────▶│ Active │ ║
║ │ 🟡 │ │ 🟢 │ ║
║ └─────┬───────┘ └─────┬───────┘ ║
║ │ │ ║
║ │ cancel │ clear ║
║ │ │ ║
║ ▼ ▼ ║
║ ┌─────────────┐ ┌─────────────┐ ║
║ │ Cleared │◀────────────────────────┤ │ ║
║ │ ⚫ │ │ │ ║
║ └─────────────┘ └─────────────┘ ║
║ ▲ ║
║ │ ║
║ │ (拒絕 rain_alert - 已清除無需暫停) ║
║ │ ║
║ ┌─────────────┐ rain_alert ┌─────────────┐ ║
║ │Emergency │◀────────────────────────┤ │ ║
║ │ Paused │ │ │ ║
║ │ 🔴 │ │ │ ║
║ └─────┬───────┘ └─────────────┘ ║
║ │ ║
║ │ resume (依 Memento 動態返回) ║
║ │ ║
║ ├──────────────┐ ║
║ │ │ ║
║ ▼ ▼ ║
║ [回到 Preparing] [回到 Active] ║
║ ║
╚══════════════════════════════════════════════════════════╝
圖例：
🟡 Preparing - 準備中
🟢 Active - 已生效
🔴 EmergencyPaused - 緊急暫停
⚫ Cleared - 已解除
事件接收 → 版本檢查 → 狀態處理 → 狀態轉移 → 副作用執行
│ │ │ │ │
▼ ▼ ▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐
│⚡ Event │ │📋 Ver │ │🎯 State │ │🔄 Trans │ │📢 Side │
│Received │ │Check │ │Handler │ │ition │ │Effects │
└─────────┘ └─────────┘ └─────────┘ └─────────┘ └─────────┘
│ │ │ │ │
│ │ (拒絕) │ │ │
│ └─────────▶│ ❌ 版本 │ │
│ │ 衝突回應 │ │
│ └─────────┘ │
│ │
└─────────────────────────────────────────▶│ 📝 審計
│ 紀錄
└─────────┘
iThome鐵人賽
看影片追技術